package com.zshield.run;

import com.zshield.config.KafkaConfig;
import com.zshield.consumer.Consumer;
import com.zshield.producer.CustomKafkaProducer;
import com.zshield.queryEs.QueryEsSendKafka;
import com.zshield.stream.Stream;
import com.zshield.util.ESclient;
import com.zshield.util.HealthMonitor;
import com.zshield.util.KafkaClusterStatus;
import com.zshield.util.MarkUpgrade;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.ParseException;
import org.apache.log4j.Logger;

import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Arrays;
import java.util.Enumeration;

public class KafkaPreCompute {
    public static String es_host;
    public static int es_number_of_replica;
    public static String bootstrap_server;
    public static String consumer_client_id;
    public static String stream_client_id;
    public static boolean is_upgrade;
    public static String input_topic;
    public static String medium_topic;
    public static String output_topic;
    public static final Logger logger = Logger.getLogger(KafkaPreCompute.class);

    public static void main(String[] args) {
        parameterParsing(args);
        KafkaClusterStatus.isClusterConnected();

        try {
            if (is_upgrade) {
                MarkUpgrade mark = new MarkUpgrade();
                if (mark.isFileExist()) {
                    while (true) {
                        logger.info("[please check if kstream upgrade is successful]");
                        Thread.sleep(1000);
                    }
                }
                KafkaClusterStatus.deleteTopic();
                String[] indexes = ESclient.listEsIndex();
                for (int i = 0; i < indexes.length; i++) {
                    if (indexes[i].startsWith("datamap")) {
                        ESclient.deleteEsIndex(indexes[i]);
                        logger.info("delete es index:[" + indexes[i] + "]");
                    }
                }
                CustomKafkaProducer producer = CustomKafkaProducer.getInstance();
                QueryEsSendKafka queryEs = new QueryEsSendKafka(producer);
                Thread t = new Thread(queryEs);
                t.start();
            }

            Stream stream = new Stream();
            Thread s = new Thread(stream, "stream");
            s.start();

            Consumer consume = new Consumer();
            Thread c = new Thread(consume, "consume");
            c.start();

            HealthMonitor.startServer(s);
        } catch (Exception e) {
            KafkaConfig.printErrorLog(logger, e);
        }
    }

    public static void parameterParsing(String[] args) {
        try {
            CommandLine cmdLine = OptionsProcessor.parseArg(args);
            CmdLineParams.setLine(cmdLine);
            es_host = CmdLineParams.getEsHost();
            es_number_of_replica = CmdLineParams.getEsNumberOfReplica();
            bootstrap_server = CmdLineParams.getBootstrapServer();
            is_upgrade = CmdLineParams.getUpgrade();
            if (is_upgrade) {
                input_topic = "sensor_input_upgrade";
                medium_topic = "sensor_output_medium";
            } else {
                input_topic = "sensor_input";
                medium_topic = "sensor_output_medium";
                output_topic = "sensor_output";
            }
            String ip = getLocalIpAddress(bootstrap_server);
            consumer_client_id = "consumer-" + ip;
            stream_client_id = "stream-" + ip;
        } catch (ParseException e1) {
            logger.error("[analytic parameter exception] " + Arrays.toString(args));
            System.exit(1);
        } catch (Exception e) {
            logger.error("[obtain local ip failure] [The reason for error {" + e + "}]");
            System.exit(1);
        }
    }

    public static String getLocalIpAddress(String bootstrap_server) throws SocketException {
        String[] serverIps;
        if (bootstrap_server.contains(",")) {
            serverIps = bootstrap_server.split(",");
            for (int i = 0; i < serverIps.length; i++) {
                serverIps[i] = serverIps[i].split(":")[0];
            }
        } else {
            serverIps = new String[1];
            serverIps[0] = bootstrap_server.split(":")[0];
        }

        for (Enumeration interfaces = NetworkInterface.getNetworkInterfaces(); interfaces.hasMoreElements(); ) {
            NetworkInterface networkInter = (NetworkInterface) interfaces.nextElement();
            for (Enumeration inetAddrs = networkInter.getInetAddresses(); inetAddrs.hasMoreElements(); ) {
                InetAddress inetAddr = (InetAddress) inetAddrs.nextElement();
                String ip = inetAddr.getHostAddress();
                for (String serverIp : serverIps) {
                    if (serverIp.equals(ip)) {
                        return serverIp;
                    }
                }
            }
        }
        return "localhost";
    }

}
